import consumer.kafka.*;
import kafka.api.FetchRequestBuilder;

import java.util.Map;
import java.util.Properties;

public class test {
    private static  String committedPath(KafkaConfig kafkaConfig) {

        return  kafkaConfig._stateConf.get(Config.ZOOKEEPER_CONSUMER_PATH) + "/"
                +  kafkaConfig._stateConf.get(Config.KAFKA_CONSUMER_ID) + "/"
                +  kafkaConfig._stateConf.get(Config.KAFKA_TOPIC) + "/" + "partition_"+0;
    }
    public static void main(String[] args) {
        /**
         * (SimpleConsumer kafkaConsumer, String topic,
         int partition, KafkaConfig config)
         */
        Properties props = new Properties();
        props.put("kafka.topic", "gizwits_raw");
        props.put("zookeeper.hosts", "s60");
        props.put("zookeeper.port", "2081");
        props.put("zookeeper.kafkaConsumer.connection", "s60:2181");
        props.put("kafka.kafkaConsumer.id", "12345");
        props.put("auto.offset.reset", "smallest");
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "10");
        //这里的leadBroker, port是配对的，应该是metadata.leader().port()
     //   SimpleConsumer kafkaConsumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);

//        KafkaUtils.getOffset(kafkaConsumer,"testkafka-topic",0,new KafkaConfig(props));
// 封装一些配置信息
        KafkaConfig kafkaConfig = new KafkaConfig(props);

        System.out.println("---");
        ZkState zkState = new ZkState(kafkaConfig);

        System.out.println("------");
       DynamicPartitionConnections connections = new DynamicPartitionConnections(kafkaConfig,
                new ZkBrokerReader(kafkaConfig, zkState));
       // connections = connections.register("s60", 0);
        DynamicBrokersReader brokersReader = new DynamicBrokersReader(kafkaConfig, zkState);
        /**
         * PartitionManager man = new PartitionManager(
         _connections,
         new ZkState((String) _config._stateConf
         .get(Config.ZOOKEEPER_CONSUMER_CONNECTION)),
         _kafkaconfig, id, _receiver, _restart);
         _managers.put(id, man);
         */

    //    Map<Object, Object> map = zkState.readJSON(committedPath(kafkaConfig));
      //  System.out.println(map);



    }
}
